package com.nulldev.util.concurrency.threadIt.v4.executors.fje;

import java.util.List;
import java.util.Queue;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Future;
import java.util.concurrent.Phaser;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

import com.nulldev.util.JVM.Arguments;
import com.nulldev.util.VariableAPI.ArrayUtils;
import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutionPreference;
import com.nulldev.util.concurrency.threadIt.v4.emapi.compat.ExecutableFuture;
import com.nulldev.util.concurrency.threadIt.v4.emapi.compat.ExecutableRunnableFuture;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExecutorFeatureExtensions;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExtendedExecutor;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.concurrency.threadIt.v4.executors.fje.interfaces.ForkJoinThreadFactory;
import com.nulldev.util.concurrency.threadIt.v4.executors.fje.interfaces.ThreadPicker;
import com.nulldev.util.concurrency.threadIt.v4.executors.fje.utils.CircularThreadPicker;
import com.nulldev.util.concurrency.threadIt.v4.executors.fje.utils.RoundRobinThreadPicker;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.TEConditions;

public class ForkJoinExecutorService extends AbstractExecutorService implements ExtendedExecutor {

	protected static final boolean USE_RR_TP = Arguments.hasArgument("--FJE.useRoundRobin");
	private final List<ForkJoinThread> threads;
	private final ThreadPicker<ForkJoinThread> threadPicker;
	private final ThreadPicker<ForkJoinThread> threadPickerLP;
	private final AtomicBoolean shutdown = new AtomicBoolean(false);
	private final AtomicInteger threadsStarted = new AtomicInteger();
	private final Phaser stopLatch = new Phaser();
	private final ReentrantLock moreWorkLock = new ReentrantLock(true);
	private final Condition moreWork = moreWorkLock.newCondition();
	private final AtomicBoolean enabledStatistics;

	public ForkJoinExecutorService(final ForkJoinThreadFactory factory, final int numThreads) {
		this(factory, "ForkJoinES-Thread-", numThreads);
	}

	public ForkJoinExecutorService(final ForkJoinThreadFactory factory, final int numThreads, final boolean enableLowPriorityThreads) {
		this(factory, "ForkJoinES-Thread-", numThreads, enableLowPriorityThreads);
	}

	public ForkJoinExecutorService(final ForkJoinThreadFactory factory, final String threadPrefix, final int numThreads) {
		this(factory, threadPrefix, numThreads, false);
	}

	public ForkJoinExecutorService(final ForkJoinThreadFactory factory, final String threadPrefix, final int numThreads,
			final boolean enableLowPriorityThreads) {
		if (factory == null)
			throw new NullPointerException("factory == null");
		else if (threadPrefix == null)
			throw new NullPointerException("threadPrefix == null!");
		else if (numThreads <= 0)
			throw new IllegalArgumentException("numThreads <= 0!");
		this.enabledStatistics = new AtomicBoolean(Boolean.parseBoolean(System.getProperty("forkjoin.enableStatistics", "false")));
		this.threads = new CopyOnWriteArrayList<ForkJoinThread>();
		for (int i = 0; i < numThreads; i++) {
			final ForkJoinThread t = factory.newThread(this);
			t.setName(threadPrefix + i);
			this.threads.add(t);
		}
		// TODO: Seperate the LOW_PRIORITY threads here and init both threadPicker AND
		// threadPickerLP.
		if (enableLowPriorityThreads) {
			final int lowPriThreads = enableLowPriorityThreads ? MathUtil.clamp(4, 32, numThreads / 4) : 0;
			final SWorkStealingThreadFactory sfactory = new SWorkStealingThreadFactory();
			for (int i = 0; i < lowPriThreads; i++) {
				final ForkJoinThread t = sfactory.newThread(this);
				t.setName("L" + threadPrefix + i);
				t.setPriority(Thread.MIN_PRIORITY);
				this.threads.add(t);
			}
			final List<ForkJoinThread> standardThreads = this.threads.subList(0, numThreads - 1);
			final List<ForkJoinThread> lpThreads = this.threads.subList(numThreads, numThreads + lowPriThreads);
			this.threadPicker = USE_RR_TP ? new RoundRobinThreadPicker<ForkJoinThread>(standardThreads)
					: new CircularThreadPicker<ForkJoinThread>(standardThreads);
			this.threadPickerLP = USE_RR_TP ? new RoundRobinThreadPicker<ForkJoinThread>(lpThreads) : new CircularThreadPicker<ForkJoinThread>(lpThreads);
		} else {
			this.threadPicker = USE_RR_TP ? new RoundRobinThreadPicker<ForkJoinThread>(this.threads) : new CircularThreadPicker<ForkJoinThread>(this.threads);
			this.threadPickerLP = null;
		}
	}

	public ForkJoinExecutorService(final ForkJoinThreadFactory factory) {
		this(factory, Runtime.getRuntime().availableProcessors() * 4);
	}

	public ForkJoinExecutorService() {
		this(new WorkStealingThreadFactory());
	}

	public ForkJoinExecutorService(final String prefix) {
		this(new WorkStealingThreadFactory(), prefix, Runtime.getRuntime().availableProcessors() * 4);
	}

	public ForkJoinExecutorService(final int numThreads) {
		this(new WorkStealingThreadFactory(), numThreads);
	}

	public ForkJoinExecutorService(final int numThreads, final boolean enableLPT) {
		this(new WorkStealingThreadFactory(), numThreads, enableLPT);
	}

	public ForkJoinExecutorService(final String prefix, final int numThreads) {
		this(new WorkStealingThreadFactory(), prefix, numThreads);
	}

	public ForkJoinExecutorService(final String prefix, final int numThreads, final boolean enableLPT) {
		this(new WorkStealingThreadFactory(), prefix, numThreads, enableLPT);
	}

	@Override
	public void execute(final Runnable command) {
		TEConditions.validRunnable(command);
		if (!shutdown.get()) {
			threadPicker.nextThread().fork(new AbstractForkJoinRunnable() {
				protected void solve() {
					command.run();
				}
			});
		} else {
			throw new RejectedExecutionException(getClass().getCanonicalName() + " is shutting down!");
		}
	}

	/**
	 * Executes the given {@link AbstractForkJoinTask} and returns the result.
	 * 
	 * @param command the {@link AbstractForkJoinTask} to be executed
	 * @return computed result of the {@link AbstractForkJoinTask}
	 * @throws RejectedExecutionException if the task is submitted after
	 *                                    {@link #shutdown()} or
	 *                                    {@link #shutdownNow()} has been called.
	 */
	public <T> T invoke(final AbstractForkJoinTask<T> command) {
		if (!shutdown.get()) {
			return threadPicker.nextThread().fork(new AbstractForkJoinTask.TerminalForkJoinTask<T>(command)).join();
		} else {
			throw new RejectedExecutionException(getClass().getCanonicalName() + " is shutting down!");
		}
	}

	@Override
	public void shutdown() {
		shutdown.set(true);
		threads.forEach(ForkJoinThread::shutdown);
	}

	@Override
	public List<Runnable> shutdownNow() {
		shutdown.set(true);
		return threads.stream().map(ForkJoinThread::shutdownNow).flatMap(List::stream).collect(Collectors.toList());
	}

	@Override
	public boolean isShutdown() {
		return shutdown.get();
	}

	@Override
	public boolean isTerminated() {
		return stopLatch.isTerminated();
	}

	@Override
	public boolean awaitTermination(final long timeout, final TimeUnit unit) throws InterruptedException {
		try {
			stopLatch.awaitAdvanceInterruptibly(0, timeout, unit);
		} catch (TimeoutException e) {
			e.printStackTrace();
		}
		return stopLatch.isTerminated();
	}

	/**
	 * Blocks indefinitely until all tasks have completed execution after a shutdown
	 * request.
	 */
	public void awaitTermination() {
		stopLatch.awaitAdvance(0);
	}

	protected List<ForkJoinThread> getThreads() {
		return threads;
	}

	protected Phaser getStopLatch() {
		return stopLatch;
	}

	/**
	 * Put thread into waiting state when there is no work to do and nothing to
	 * steal. When more work is available, the thread can be notified by calling
	 * signalMoreWork()
	 * 
	 * @throws InterruptedException if the thread is interrupted.
	 * @see #signalMoreWork()
	 */
	public void awaitMoreWork() throws InterruptedException {
		moreWorkLock.lock();
		try {
			moreWork.await();
		} finally {
			moreWorkLock.unlock();
		}
	}

	/**
	 * Signals that a task has been forked onto a work queue. Waiting threads are
	 * woken up. If there are non-started threads, then one is chosen to start.
	 * 
	 * @see #awaitMoreWork()
	 */
	public void signalMoreWork() {
		if (threadsStarted.get() < threads.size()) {
			threads.get(threadsStarted.getAndIncrement()).start();
		} else {
			moreWorkLock.lock();
			try {
				moreWork.signalAll();
			} finally {
				moreWorkLock.unlock();
			}
		}
	}

	public boolean isStatisticsEnabled() {
		return enabledStatistics.get();
	}

	@Override
	public String toString() {
		return "FJE@" + MathUtil.toHex(super.hashCode()) + "[threads=" + this.threads.size() + ",started=" + threadsStarted.get() + "]";
	}

	private static final ExecutorFeatureExtensions[] SUPPORTED = new ExecutorFeatureExtensions[]
		{ ExecutorFeatureExtensions.EEX_executor_native_batching, ExecutorFeatureExtensions.EEX_core_submit_all_tasks,
				ExecutorFeatureExtensions.EEX_core_drain_task_queue, ExecutorFeatureExtensions.EEX_core_execute_with_preference,
				ExecutorFeatureExtensions.EEX_query_is_idle, ExecutorFeatureExtensions.EEX_query_batching_supported,
				ExecutorFeatureExtensions.EEX_query_slots_supported, ExecutorFeatureExtensions.EEX_query_thread_affinity_supported,
				ExecutorFeatureExtensions.EEX_executor_low_priority_threads, ExecutorFeatureExtensions.EEX_query_executor_forking,
				ExecutorFeatureExtensions.EEX_compat_native_concurrency_api, ExecutorFeatureExtensions.EEX_query_debug, };

	@Override
	public boolean supportsFeature(final ExecutorFeatureExtensions feature) {
		return ArrayUtils.contains(SUPPORTED, feature);
	}

	@Override
	public ExecutorFeatureExtensions[] supported() {
		return SUPPORTED;
	}

	@Override
	public void execute(final Runnable r, final ExecutionPreference e) {
		TEConditions.validRunnable(r);
		this.execute(r);
	}

	@Override
	public boolean executorIsIdle() {
		return !this.moreWorkLock.isLocked();
	}

	/* EEX_query_batching_supported */

	@Override
	public boolean executorSupportsBatching() {
		return true;
	}

	/* EEX_query_slots_supported */

	@Override
	public boolean executorSupportsSlotCount() {
		return false;
	}

	/* */

	@Override
	public void batch(final Runnable f) {
		TEConditions.validRunnable(f);
		if (!shutdown.get()) {
			threadPicker.nextThread().fork(new AbstractForkJoinRunnable() {
				protected void solve() {
					f.run();
				}
			});
		} else {
			throw new RejectedExecutionException(getClass().getCanonicalName() + " is shutting down!");
		}
	}

	@Override
	public void batchAll(final Runnable[] _r) {
		if (_r == null || _r.length <= 0)
			return;
		if (!shutdown.get()) {
			for (int index = 0; index < _r.length; index++) {
				final Runnable r = _r[index];
				threadPicker.nextThread().fork(new AbstractForkJoinRunnable() {
					protected void solve() {
						r.run();
					}
				});
			}
		} else {
			throw new RejectedExecutionException(getClass().getCanonicalName() + " is shutting down!");
		}
	}

	@Override
	public void executor_submitAll(final List<Runnable> _r) {
		if (_r == null || _r.isEmpty())
			return;
		if (!shutdown.get()) {
			final List<ForkJoinThread> ts = threadPicker.nextThreads(_r.size());
			for (int i = 0; i < _r.size(); i++) {
				final Runnable r = _r.get(i);
				if (r == null)
					continue;
				ts.get(i).fork(new AbstractForkJoinRunnable() {
					protected void solve() {
						r.run();
					}
				});
			}
		} else {
			throw new RejectedExecutionException(getClass().getCanonicalName() + " is shutting down!");
		}
	}

	@Override
	public void executor_drainTasks(final Queue<Runnable> async_queue, final int batchSize) {
		if (async_queue == null || async_queue.isEmpty())
			return;
		if (batchSize <= 0)
			return;
		if (!shutdown.get()) {
			final List<ForkJoinThread> ts = threadPicker.nextThreads(batchSize);
			for (int i = 0; i < batchSize; i++) {
				final Runnable r = async_queue.poll();
				if (r == null)
					continue;
				ts.get(i).fork(new AbstractForkJoinRunnable() {
					protected void solve() {
						r.run();
					}
				});
			}
		} else {
			throw new RejectedExecutionException(getClass().getCanonicalName() + " is shutting down!");
		}
	}

	/* EEX_query_thread_affinity_supported */

	@Override
	public boolean executorSupportsThreadAffinity() {
		return false;
	}

	@Override
	public boolean executorSupportsThreadPinning() {
		return false;
	}

	/* EEX_executor_low_priority_threads */

	@Override
	public boolean supportsLPT() {
		return true;
	}

	@Override
	public boolean isLPTActive() {
		return this.threadPickerLP != null;
	}

	@Override
	public void setLPTActive(final boolean state) {
		this.unsupported(ExecutorFeatureExtensions.EEX_executor_low_priority_threads);
	}

	@Override
	public int[] getLPTs() {
		this.unsupported(ExecutorFeatureExtensions.EEX_executor_low_priority_threads);
		return null;
	}

	@Override
	public void submitLP(final Runnable command) {
		TEConditions.validRunnable(command);
		if (this.threadPickerLP == null) {
			this.submit(command);
			return;
		}
		if (!shutdown.get()) {
			threadPickerLP.nextThread().fork(new AbstractForkJoinRunnable() {
				protected void solve() {
					command.run();
				}
			});
		} else {
			throw new RejectedExecutionException(getClass().getCanonicalName() + " is shutting down!");
		}
	}

	@Override
	public void submitLP(final Executable<?> r) {
		TEConditions.validExecutable(r);
		if (r instanceof Runnable) {
			this.submitLP((Runnable) r);
		} else {
			this.submitLP(() -> {
				r.run_st();
			});
		}
	}

	/* EEX_query_executor_forking */

	@Override
	public boolean executorIsForking() {
		return true;
	}

	/* EEX_compat_native_concurrency_api */

	@Override
	public Future<?> submit(final Runnable task) {
		TEConditions.validRunnable(task);
		final ExecutableRunnableFuture<Object> ef = new ExecutableRunnableFuture<Object>(task, null);
		execute(ef);
		return ef;
	}

	@Override
	public <T> Future<T> submit(final Runnable task, final T result) {
		TEConditions.validRunnable(task);
		final ExecutableRunnableFuture<T> ef = new ExecutableRunnableFuture<T>(task, result);
		execute(ef);
		return ef;
	}

	@Override
	public <T> Future<T> submit(final Callable<T> task) {
		final ExecutableFuture<T> ef = new ExecutableFuture<T>(task);
		execute(ef);
		return ef;
	}

	@Override
	protected <T> RunnableFuture<T> newTaskFor(final Callable<T> callable) {
		return new ExecutableFuture<T>(callable);
	}

	@Override
	protected <T> RunnableFuture<T> newTaskFor(final Runnable runnable, final T value) {
		TEConditions.validRunnable(runnable);
		return new ExecutableRunnableFuture<T>(runnable, value);
	}

	/* EEX_query_debug */

	@Override
	public Object debug_query(final EEXDebugQueryType type) {
		switch (type) {
			case EEX_DEBUG_ACTIVE_THREAD_COUNT:
			case EEX_DEBUG_TOTAL_THREAD_COUNT:
				return this.getThreads().size();
			case EEX_DEBUG_TASKS_IN_QUEUE:
				return 0; /* FJE doesn't have the concept of queuing */
			/* */
			default:
				return null;
		}
	}
}